package com.morihacky.android.rxjava;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;

import java.util.ArrayList;
import java.util.List;

import butterknife.ButterKnife;
import butterknife.OnClick;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

public class Main2Activity extends AppCompatActivity {

	public static final String TAG = "itchen-->";

	@Override
	protected void onCreate(Bundle savedInstanceState) {
		super.onCreate(savedInstanceState);
		setContentView(R.layout.activity_main2);

		Log.d(TAG, Thread.currentThread().getName());

		ButterKnife.bind(this);


	}

	@OnClick(R.id.buttonid8)
	void OnButton8(){

		//这个不是转换,其实是拆分.拆开,将大的模块拆分成小模块发送消息.
		Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				emitter.onNext(1);
				emitter.onNext(2);
				emitter.onNext(3);
				emitter.onComplete();
			}
			//其实这个不是转换,是拆分.
		}).flatMap(new Function<Integer, ObservableSource<String>>() {
			@Override
			public ObservableSource<String> apply(Integer integer) throws Exception {

				//转换操作.
				final List<String> list = new ArrayList<>();
				for (int i = 0; i < 3; i++) {
					list.add("I am value " + integer);
				}



				return null;
			}
		}).subscribe(new Consumer<String>() {


			@Override
			public void accept(String s) throws Exception {

			}
		});


	}

	@OnClick(R.id.buttonid7)
	void OnButton7() {

		Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> observableemitter) throws Exception {

				observableemitter.onNext(1);
				observableemitter.onNext(2);
				observableemitter.onNext(3);
				observableemitter.onComplete();

			}
		}).map(new Function<Integer, String>() {
			@Override
			public String apply(Integer integer) throws Exception {
				return "itchen---This is result " + integer;
			}
		}).subscribe(new Consumer<String>() {
			@Override
			public void accept(String s) throws Exception {

				//处理最终的结果.
				Log.d(TAG, "处理结果-->" + s);

			}
		});

		//02-20 14:21:22.931 27382-27382/com.morihacky.android.rxjava D/itchen-->: main
		//02-20 14:21:24.501 27382-27382/com.morihacky.android.rxjava D/itchen-->: 处理结果-->This is result 1
		//02-20 14:21:24.501 27382-27382/com.morihacky.android.rxjava D/itchen-->: 处理结果-->This is result 2
		//02-20 14:21:24.501 27382-27382/com.morihacky.android.rxjava D/itchen-->: 处理结果-->This is result 3

	}


	@OnClick(R.id.buttonid6)
	void OnButton6() {

		Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
				Log.d(TAG, "emit 1");
				emitter.onNext(1);
			}
		});

		Consumer<Integer> consumer = new Consumer<Integer>() {
			@Override
			public void accept(Integer integer) throws Exception {
				Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
				Log.d(TAG, "onNext: " + integer);
			}
		};


		//第一次测试只指定一次:
		/*observable.subscribeOn(Schedulers.newThread())//子线程中发送数据(订阅发生在子线程或者被观察者在子线程发射数据).//上游发送事件的线程.
				.observeOn(AndroidSchedulers.mainThread())//主线程中处理数据(观察)接收事件处理数据所在的线程.
				.subscribe(consumer);*/
		//		02-06 17:16:23.228 24952-24952/com.morihacky.android.rxjava D/itchen-->: main
		//		02-06 17:16:24.988 24952-28111/com.morihacky.android.rxjava D/itchen-->: Observable thread is : RxNewThreadScheduler-1
		//		02-06 17:16:24.988 24952-28111/com.morihacky.android.rxjava D/itchen-->: emit 1
		//		02-06 17:16:24.998 24952-24952/com.morihacky.android.rxjava D/itchen-->: Observer thread is :main
		//		02-06 17:16:24.998 24952-24952/com.morihacky.android.rxjava D/itchen-->: onNext: 1

		//第二次测试,多次指定:
		/*observable.subscribeOn(Schedulers.newThread())//在子线程发送-----itchen结果是以第一次为准.
				.subscribeOn(Schedulers.io())//切换到主线程发送
				.observeOn(AndroidSchedulers.mainThread())//主线程处理
				.observeOn(Schedulers.io())//io线程处理---itchen结果是以最后一次为准.
				.subscribe(consumer);//订阅关系产生*/

		//		02-06 17:36:26.428 12202-12686/com.morihacky.android.rxjava D/itchen-->: Observable thread is : RxNewThreadScheduler-1
		//		02-06 17:36:26.428 12202-12686/com.morihacky.android.rxjava D/itchen-->: emit 1
		//		02-06 17:36:26.438 12202-12687/com.morihacky.android.rxjava D/itchen-->: Observer thread is :RxCachedThreadScheduler-2
		//		02-06 17:36:26.438 12202-12687/com.morihacky.android.rxjava D/itchen-->: onNext: 1

		//第三次测试,添加日志,仔细打印日志内容
		observable.subscribeOn(Schedulers.newThread())
				.subscribeOn(Schedulers.io())
				.observeOn(AndroidSchedulers.mainThread())
				.doOnNext(new Consumer<Integer>() {
					@Override
					public void accept(Integer integer) throws Exception {
						Log.d(TAG, "After observeOn(mainThread), current thread is: " + Thread.currentThread().getName());
					}
				})
				.observeOn(Schedulers.io())
				.doOnNext(new Consumer<Integer>() {
					@Override
					public void accept(Integer integer) throws Exception {
						Log.d(TAG, "After observeOn(io), current thread is : " + Thread.currentThread().getName());
					}
				})
				.subscribe(consumer);

		//		02-06 17:40:22.018 16057-16360/com.morihacky.android.rxjava D/itchen-->: Observable thread is : RxNewThreadScheduler-1
		//		02-06 17:40:22.018 16057-16360/com.morihacky.android.rxjava D/itchen-->: emit 1
		//		02-06 17:40:22.028 16057-16057/com.morihacky.android.rxjava D/itchen-->: After observeOn(mainThread), current thread is: main
		//		02-06 17:40:22.028 16057-16361/com.morihacky.android.rxjava D/itchen-->: After observeOn(io), current thread is : RxCachedThreadScheduler-2
		//		02-06 17:40:22.028 16057-16361/com.morihacky.android.rxjava D/itchen-->: Observer thread is :RxCachedThreadScheduler-2
		//		02-06 17:40:22.028 16057-16361/com.morihacky.android.rxjava D/itchen-->: onNext: 1

	}

	@OnClick(R.id.buttonid5)
	void OnButton5() {

		//被观察者.
		Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
				Log.d(TAG, "emit 1");
				emitter.onNext(1);
			}
		});

		//数据观察者.
		Consumer<Integer> consumer = new Consumer<Integer>() {
			@Override
			public void accept(Integer integer) throws Exception {
				Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
				Log.d(TAG, "onNext: " + integer);
			}
		};

		//产生订阅.
		observable.subscribe(consumer);

		//		02-06 17:01:23.548 14049-14049/com.morihacky.android.rxjava D/itchen-->: main
		//		02-06 17:01:24.898 14049-14049/com.morihacky.android.rxjava D/itchen-->: Observable thread is : main
		//		02-06 17:01:24.898 14049-14049/com.morihacky.android.rxjava D/itchen-->: emit 1
		//		02-06 17:01:24.898 14049-14049/com.morihacky.android.rxjava D/itchen-->: Observer thread is :main
		//		02-06 17:01:24.898 14049-14049/com.morihacky.android.rxjava D/itchen-->: onNext: 1

	}

	@OnClick(R.id.buttonid4)
	void OnButton() {

		Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				Log.d(TAG, "emit 1");
				emitter.onNext(1);
				Log.d(TAG, "emit 2");
				emitter.onNext(2);
				Log.d(TAG, "emit 3");
				emitter.onNext(3);
				Log.d(TAG, "emit complete");
				emitter.onComplete();
				Log.d(TAG, "emit 4");
				emitter.onNext(4);
			}
		}).subscribe(new Consumer<Integer>() {
			@Override
			public void accept(Integer integer) throws Exception {
				Log.d(TAG, "onNext: " + integer);//只接受关心的内容.
			}
		});

		//		02-06 16:50:39.578 4047-4047/com.morihacky.android.rxjava D/itchen: emit 1
		//		02-06 16:50:39.578 4047-4047/com.morihacky.android.rxjava D/itchen: onNext: 1
		//		02-06 16:50:39.578 4047-4047/com.morihacky.android.rxjava D/itchen: emit 2
		//		02-06 16:50:39.578 4047-4047/com.morihacky.android.rxjava D/itchen: onNext: 2
		//		02-06 16:50:39.578 4047-4047/com.morihacky.android.rxjava D/itchen: emit 3
		//		02-06 16:50:39.578 4047-4047/com.morihacky.android.rxjava D/itchen: onNext: 3
		//		02-06 16:50:39.578 4047-4047/com.morihacky.android.rxjava D/itchen: emit complete
		//		02-06 16:50:39.578 4047-4047/com.morihacky.android.rxjava D/itchen: emit 4

	}

	@OnClick(R.id.buttonid3)
	void OnButton3() {

		Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				Log.d(TAG, "emit 1");
				emitter.onNext(1);
				Log.d(TAG, "emit 2");
				emitter.onNext(2);
				Log.d(TAG, "emit 3");
				emitter.onNext(3);
				Log.d(TAG, "emit complete");
				emitter.onComplete();
				Log.d(TAG, "emit 4");
				emitter.onNext(4);
			}
		}).subscribe(new Observer<Integer>() {
			private Disposable mDisposable;
			private int i;

			@Override
			public void onSubscribe(Disposable d) {
				Log.d(TAG, "subscribe");
				mDisposable = d;
			}

			@Override
			public void onNext(Integer value) {
				Log.d(TAG, "onNext: " + value);
				i++;
				if (i == 2) {
					Log.d(TAG, "dispose");
					mDisposable.dispose();
					Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
				}
			}

			@Override
			public void onError(Throwable e) {
				Log.d(TAG, "error");
			}

			@Override
			public void onComplete() {
				Log.d(TAG, "complete");
			}
		});

		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: subscribe
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: emit 1
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: onNext: 1
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: emit 2
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: onNext: 2
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: dispose
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: isDisposed : true
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: emit 3
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: emit complete
		//		02-06 16:32:41.168 19408-19408/com.morihacky.android.rxjava D/itchen: emit 4

	}

	@OnClick(R.id.buttonid2)
	void onButton2() {


		Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> e) throws Exception {
				e.onNext(1);
				e.onNext(2);
				e.onNext(3);
				e.onComplete();
			}
		}).subscribe(new Observer<Integer>() {
			@Override
			public void onSubscribe(Disposable d) {
				LogUtils.d("itchen--onSubscribe-->" + d);
			}

			@Override
			public void onNext(Integer value) {
				LogUtils.d("itchen--onNext-->" + value);
			}

			@Override
			public void onError(Throwable e) {
				LogUtils.d("itchen--onError-->" + e);
			}

			@Override
			public void onComplete() {
				LogUtils.d("itchen--onComplete-->");
			}
		});
	}

	@OnClick(R.id.buttonid1)
	void onButton() {

		//创建一个上游 Observable：
		/*Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				emitter.onNext(1);
				emitter.onNext(2);
				emitter.onNext(3);
				emitter.onComplete();
			}
		});*/
		//创建一个下游 Observer
		/*Observer<Integer> observer = new Observer<Integer>() {
			@Override
			public void onSubscribe(Disposable d) {
				//Log.d(TAG, "subscribe");
				LogUtils.d("itchen--onSubscribe-->"+d);
			}

			@Override
			public void onNext(Integer value) {
				//Log.d(TAG, "" + value);
				LogUtils.d("itchen--onNext-->"+value);
			}

			@Override
			public void onError(Throwable e) {
				//Log.d(TAG, "error");
			}

			@Override
			public void onComplete() {
				//Log.d(TAG, "complete");
				LogUtils.d("itchen--onComplete-->");
			}
		};*/

		//建立连接
		//observable.subscribe(observer);

		//自己手动写上边的例子==================================================================
		//创建一个被观察者,数据发射器,
		io.reactivex.Observable<Integer> observable = io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				emitter.onNext(1);
				emitter.onNext(2);
				emitter.onNext(3);
				emitter.onComplete();
			}
		});

		//创建一个下游
		Observer<Integer> observer = new Observer<Integer>() {
			@Override
			public void onSubscribe(Disposable d) {
				LogUtils.d("itchen--onSubscribe-->" + d);
			}

			@Override
			public void onNext(Integer value) {
				LogUtils.d("itchen--onNext-->" + value);
			}

			@Override
			public void onError(Throwable e) {
				LogUtils.d("itchen--onError-->" + e);
			}

			@Override
			public void onComplete() {
				LogUtils.d("itchen--onComplete-->");
			}
		};

		//被观察者订阅观察者.
		observable.subscribe(observer);

		//		02-06 14:09:35.388 9920-9920/com.morihacky.android.rxjava D/Main2Activity$2.onSubscribe(L:90): itchen--onSubscribe-->null
		//		02-06 14:09:35.388 9920-9920/com.morihacky.android.rxjava D/Main2Activity$2.onNext(L:95): itchen--onNext-->1
		//		02-06 14:09:35.398 9920-9920/com.morihacky.android.rxjava D/Main2Activity$2.onNext(L:95): itchen--onNext-->2
		//		02-06 14:09:35.398 9920-9920/com.morihacky.android.rxjava D/Main2Activity$2.onNext(L:95): itchen--onNext-->3
		//		02-06 14:09:35.398 9920-9920/com.morihacky.android.rxjava D/Main2Activity$2.onComplete(L:105): itchen--onComplete-->


	}
}
